package com.simple.mqtt.common.task;

import com.simple.mqtt.common.tool.LogTool;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.mqtt.MqttMessage;
import io.netty.handler.codec.mqtt.MqttPublishMessage;
import io.netty.handler.codec.mqtt.MqttQoS;

import java.util.concurrent.ScheduledExecutorService;

/**
 * Handles MQTT message publish retransmission mechanism
 * Supports QoS 1 and QoS 2 level message acknowledgment and retransmission
 */
public final class MqttPendingPublish extends AbstractMqttPendingMessage<MqttPublishMessage> {
    /** Original publish message */
    private final MqttPublishMessage message;
    /** PUBREL message retransmission processor (for QoS 2) */
    private final RetryProcessor<MqttMessage> pubRelRetryProcessor = new RetryProcessor<>();

    /**
     * Constructor
     * @param message publish message
     * @param qos quality of service level
     */
    public MqttPendingPublish(MqttPublishMessage message, MqttQoS qos) {
        super(qos, new RetryProcessor<>());
        this.message = message;
        this.retryProcessor.setOriginalMessage(message);
    }

    /**
     * Start publish message retransmission timer
     * @param taskService scheduled task service
     * @param context channel context
     */
    public void startPublishRetransmissionTimer(ScheduledExecutorService taskService, ChannelHandlerContext context) {
        startRetransmissionTimer(taskService, context,
                (header, msg) -> new MqttPublishMessage(header, msg.variableHeader(), msg.payload()),
                "Publish");
    }

    /**
     * Called when PUBACK acknowledgment message is received
     * Stops retransmission timer
     */
    public void onPubAckReceived() {
        stopRetransmission();
    }

    /**
     * Set PUBREL message
     * @param pubRelMessage PUBREL message object
     */
    public void setPubRelMessage(MqttMessage pubRelMessage) {
        this.pubRelRetryProcessor.setOriginalMessage(pubRelMessage);
    }

    public void startPubRelRetransmissionTimer(ScheduledExecutorService taskService, ChannelHandlerContext context) {
        pubRelRetryProcessor.setHandler((fixedHeader, originalMessage) -> {
            MqttMessage pubRelMessage = new MqttMessage(fixedHeader, originalMessage.variableHeader());
            context.writeAndFlush(pubRelMessage).addListener(future -> {
                if (context.channel().isActive()) {
                    if (context.channel().isWritable()) {
                        LogTool.info("retry send PubRel msg qos:{} result:{}", qos, future.isSuccess());
                    } else {
                        LogTool.warn("retry send PubRel msg qos:{} failed, channel not writable", qos);
                    }
                } else {
                    LogTool.warn("retry send PubRel msg qos:{} failed, channel inactive", qos);
                }
            });
        });
        pubRelRetryProcessor.start(taskService);
    }

    /**
     * Called when PUBCOMP acknowledgment message is received (completes QoS 2 flow)
     * Stops PUBREL message retransmission timer
     */
    public void onPubCompReceived() {
        pubRelRetryProcessor.stop();
    }
}